package com.leyantech.ecp.agenx.dispatcher.consumer;

import com.leyantech.chaos.kafka.KafkaConsumerStringProcessor;
import com.leyantech.ecp.agenx.common.config.MessageAddress;
import com.leyantech.ecp.agenx.dispatcher.codec.ClientMessageEncoder;
import com.leyantech.ecp.agenx.dispatcher.sender.EventBusSender;

import com.google.inject.Inject;
import com.google.inject.Singleton;
import io.vertx.core.json.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-01.
 */
@Slf4j
@Singleton
public class OutPutMessageConsumer implements KafkaConsumerStringProcessor {

  private final EventBusSender sender;

  private final ClientMessageEncoder clientMessageEncoder;

  @Inject
  public OutPutMessageConsumer(EventBusSender sender,
      ClientMessageEncoder clientMessageEncoder) {
    this.sender = sender;
    this.clientMessageEncoder = clientMessageEncoder;
  }

  @Override
  public void accept(ConsumerRecord<String, String> consumerRecord) {
    log.info("received output message {}", consumerRecord);

    final JsonObject message = new JsonObject(consumerRecord.value());

    clientMessageEncoder.encode(message);

    final String recipient = consumerRecord.key();

    sender.send(MessageAddress.toConnector(recipient), message.toString());
  }

}
